[Kinesis Data Streams] ストリームを挟んでS3に一覧ログを記録してみました

SIN

2019.11.04

この記事は公開されてから1年以上経過しています。情報が古い可能性がありますので、ご注意ください。

1 はじめに

CX事業本部の平内（SIN）です。

処理ごとに蓄積される「一覧（ログ）ファイルを、リアルタイムにS3上に作成する」というニーズに対して、昨日、下記のようなブログを書きました。

これに対して、社内から、このような要件なら、「FIFOなSQSより、KinesisでSequenceで管理した方がLambdaも入れやすいし使い勝手は良さそうな気がします。」という貴重なアドバイスを頂いたので、やってみました。

注意：今回も、S3の「結果整合性」の制約下で完全な「排他制御」が出来たわけではありませんので・・・「それでも！」という方は、読み進めて頂ければ嬉しいです。

2 Kinesis Data Streams

Kinesis Data Streamsを log-sample-stream という名前で作成しました。

ストリームを処理するタスクの同時実行を制限するために、シャードを1に設定しています。ストリームの容量は、シャードの数によって決まるため、今回は、下記の読み書きの制限内で使用可能という事になります。

3 プロデューサー

プロデューサー側のコードは、以下のとおりです。こころなしか、SQSに送るより簡単になっている気がします。

import * as AWS from 'aws-sdk';

exports.handler = async (event: any) => {
    await log(event.message);
}

async function log(message: string) {
    const body = {
        message: message
    }

    const kinesis = new AWS.Kinesis();
    const streamName = 'log-sample-stream'; 
    const partitionKey = "log-sample";

    const params = {
        Data: JSON.stringify(body),
        PartitionKey: partitionKey,
        StreamName: streamName,
    };
    await kinesis.putRecord(params).promise();
}

4 コンシューマー

コンシューマー側は、次のとおりです。 Kinesisのコンシューマーは、ストリームをトリガーにできるため、パラメーターで渡されるレコードを処理しているだけです。こちらも、明らかにSQSのキューを処理するより簡単です。

import * as AWS from 'aws-sdk';

const bucketName = 'my-bucket-name';
const key = 'log.txt';

exports.handler = async (event: any) => {

    console.log(JSON.stringify(event));

    try{
        // S3のログファイルを読み込む
        const s3 = new AWS.S3();
        const s3_data = await s3.getObject( {Bucket: bucketName, Key: key} ).promise();
        let text = s3_data.Body!.toString("utf-8");

        await Promise.all(event.Records.map( (record:any) => {
            const data = Buffer.from(record.kinesis.data as string, 'base64').toString();
            console.log('data:', data);
            const body: {message:string} = JSON.parse(data);
            // ログを１行追加
            text += `${getDateTimeString()},${body.message}\n`;

        }))

        // S3へログファイルを保存する
        await s3.putObject({Bucket: bucketName, Key: key, Body: text} ).promise();
    } catch(err){
        console.log(`ERROR: ${JSON.stringify(event)}`)
    }
}

function getDateTimeString(){
    const dt = new Date();
    const Y = dt.getFullYear();
    const M = ("00" + (dt.getMonth()+1)).slice(-2);
    const D = ("00" + dt.getDate()).slice(-2);
    const h = ("00" + (dt.getHours())).slice(-2);
    const m = ("00" + (dt.getMinutes())).slice(-2);
    const s = ("00" + (dt.getSeconds())).slice(-2);
    const ms = ("000" + (dt.getMilliseconds())).slice(-3);
    const result = `${Y}/${M}/${D} ${h}:${m}:${s}.${ms}`;
    return result;
}

Kinesisをトリガーに設定しているようすです。

5 非同期処理

前回と同様に、非同期でプロデューサーを実行してみました。

const exec = require('child_process').exec;
const profile = 'my-profile';
const functionName = 'log-sample';

const max = 100;

[...Array(max).keys()].forEach( i => {
    const param = {message:i};
    exec(`aws lambda invoke --invocation-type Event --function-name ${functionName} --payload '${JSON.stringify(param)}' --p ${profile} response.json`);
})

コンシューマーのログを見ると、以下のように、複数回に分けて起動されていることが確認できました。

14件
5件
13件
6件
...略

予想外（？）に、一覧ログは、欠けること無く100件全部取得できていました。

2019/11/03 22:24:19.223,0
2019/11/03 22:24:19.281,6
2019/11/03 22:24:19.282,1
2019/11/03 22:24:19.282,12
2019/11/03 22:24:19.282,7
2019/11/03 22:24:19.282,2
2019/11/03 22:24:19.282,3
2019/11/03 22:24:19.282,8
2019/11/03 22:24:19.282,11
2019/11/03 22:24:19.301,13
2019/11/03 22:24:19.301,17
2019/11/03 22:24:19.301,9
2019/11/03 22:24:19.302,16
...略